package com.mask.common.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
@EnableConfigurationProperties(RabbitMQProperties.class)
@ConditionalOnProperty(prefix = "mask.mq", name = "enabled", havingValue = "true", matchIfMissing = true)
public class RabbitMQConfig {
    @Autowired
    private RabbitMQProperties properties;

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        
        // 消息发送失败返回到队列中
        rabbitTemplate.setMandatory(true);
        
        // 消息发送到交换机确认机制
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送到交换机失败：correlationData={}, cause={}", correlationData, cause);
            }
        });
        
        // 消息从交换机发送到队列确认机制
//        rabbitTemplate.setReturnsCallback(returned -> {
//            log.error("消息从交换机发送到队列失败：message={}, replyCode={}, replyText={}, exchange={}, routingKey={}",
//                returned.getMessage(), returned.getReplyCode(), returned.getReplyText(),
//                returned.getExchange(), returned.getRoutingKey());
//        });
        
        return rabbitTemplate;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        
        // 创建交换机
        properties.getExchanges().forEach((key, exchange) -> {
            AbstractExchange abstractExchange = createExchange(exchange);
            rabbitAdmin.declareExchange(abstractExchange);
        });
        
        // 创建队列并绑定
        properties.getQueues().forEach((key, queue) -> {
            Queue amqpQueue = new Queue(queue.getName(), queue.isDurable());
            rabbitAdmin.declareQueue(amqpQueue);
            
            // 绑定队列到交换机
            if (StringUtils.isNotEmpty(queue.getExchange())) {
                Binding binding = BindingBuilder.bind(amqpQueue)
                    .to(new DirectExchange(queue.getExchange()))
                    .with(queue.getRoutingKey());
                rabbitAdmin.declareBinding(binding);
            }
        });
        
        return rabbitAdmin;
    }

    private AbstractExchange createExchange(RabbitMQProperties.Exchange exchange) {
        switch (exchange.getType().toLowerCase()) {
            case "direct":
                return new DirectExchange(exchange.getName(), exchange.isDurable(), false);
            case "fanout":
                return new FanoutExchange(exchange.getName(), exchange.isDurable(), false);
            case "topic":
                return new TopicExchange(exchange.getName(), exchange.isDurable(), false);
            default:
                throw new IllegalArgumentException("Unsupported exchange type: " + exchange.getType());
        }
    }
}